package top.wshape1.rpc.consumer;

import io.netty.channel.Channel;
import io.netty.util.concurrent.DefaultPromise;
import lombok.extern.slf4j.Slf4j;
import top.wshape1.rpc.common.constant.RpcMessageCodeEnum;
import top.wshape1.rpc.common.entity.ServiceMeta;
import top.wshape1.rpc.common.exception.RpcException;
import top.wshape1.rpc.config.JJRpcConfig;
import top.wshape1.rpc.filter.FilterChain;
import top.wshape1.rpc.filter.FilterChainFactory;
import top.wshape1.rpc.loadbalance.LoadBalanceFactory;
import top.wshape1.rpc.message.RpcRequest;
import top.wshape1.rpc.registry.RpcRegistryFactory;
import top.wshape1.rpc.consumer.tolerant.FaultTolerantStrategyFactory;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.net.ConnectException;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

/**
 * @author Wshape1
 * @version 1.0
 * @description
 * @since 2024-02-13
 */

@Slf4j
public class RpcProxyInvocationHandler implements InvocationHandler {

    private final Class<?> serviceClass;
    private final RpcConsumer consumer;

    public RpcProxyInvocationHandler(Class<?> serviceClass, RpcConsumer consumer) {
        this.serviceClass = serviceClass;
        this.consumer = consumer;
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) {
        String serviceName = serviceClass.getName();
        RpcRequest rpcRequest = new RpcRequest(
                serviceName,
                method.getName(),
                method.getReturnType(),
                method.getParameterTypes(),
                args
        );
        rpcRequest.setSequenceId(SequenceIdGenerator.nextId());

        // 过滤器链
        FilterChain filterChain = FilterChainFactory.getConsumerSendReqFilterChain();
        try {
            if (!filterChain.doFilter(rpcRequest)) {
                log.debug("RpcConsumer调用方法：{}.{}({}), 过滤器链返回false.", serviceName, method.getName(), args);
                throw new RpcException("doFilter return false.");
            }
        } finally {
            filterChain.clearLocal();
        }


        List<ServiceMeta> discover = RpcRegistryFactory.getRegistry().lookup(serviceName);

        if (discover == null || discover.isEmpty()) {
            // 未发现可用的服务
            throw new RpcException(RpcMessageCodeEnum.RPC_SERVICE_NOT_AVAILABLE);
        }

        // 负载均衡
        ServiceMeta serviceMeta = LoadBalanceFactory.getLoadBalance().select(discover);
        Function<ServiceMeta, Object> invoker = (sm) -> {
            try {
                return getResult(sm, rpcRequest);
            } catch (ConnectException | InterruptedException e) {
                throw new RuntimeException(e);
            }
        };
        try {
            log.debug("RpcConsumer调用方法：{}.{}({}), 生产者为{}",
                    serviceName, method.getName(), args, serviceMeta.getAddress());

            return invoker.apply(serviceMeta);
        } catch (Exception e) {
            log.debug("容错机制启动！{}启动！", FaultTolerantStrategyFactory.getStrategy().getClass().getName());
            return FaultTolerantStrategyFactory.getStrategy().handle(invoker, serviceMeta, e);
        }
    }

    private Object getResult(ServiceMeta serviceMeta, RpcRequest rpcRequest) throws ConnectException, InterruptedException {
        // 去连接这个服务
        Channel channel = this.consumer.connect(serviceMeta.getIp(), serviceMeta.getPort());

        // send msg
        channel.writeAndFlush(rpcRequest);

        // 准备一个空 Promise 对象，来接收结果             指定 promise 对象异步接收结果线程
        DefaultPromise<Object> promise = new DefaultPromise<>(channel.eventLoop());
        CallFutureHolder.put(rpcRequest.getSequenceId(), promise);

        // 等待 promise 结果
        if (!promise.await(JJRpcConfig.CONFIG.getConsumer().getCallTimeout(), TimeUnit.MILLISECONDS)) {
            CallFutureHolder.remove(rpcRequest.getSequenceId());
            throw new RpcException(RpcMessageCodeEnum.RPC_TIMEOUT);
        }
        if (promise.isSuccess()) {
            // 调用正常
            return promise.getNow();
        } else {
            // 调用失败
            throw new RuntimeException(promise.cause());
        }
    }
}
